package clprotocol;

# This module contains a number of routines that the main cluser
# daemon may call on to aid various changes to the cluster configuration,
# including;
#
# 1. Cluster formation [Or optionally joining if already running]
# 2. Handling loss of existing master node
#
# The 'master' node is a node responsible for sending out certain
# events to the other nodes in the cluster.
#
# CLUSTER FORMATION DETAILS
# =========================
# The actions below will take place on all nodes that are attempting to
# form the cluster simultaneously. Various scenarios are given sicne 
# attempting to explain all logic at the send time would become unweildly.
#
#   Initialisation
#   --------------
#   Each Node will generate a series of client channels to send out
#   messages. In bcast_threshold+ nodes will be 1 channel per defined network,
#   otherise separate channels for each node/network combination.
#   [Sample code taken from lha_hbd configuration]
#
#   Scenario 1 - no nodes currently running cluster
#   -----------------------------------------------
#   Nodes send outs (sendonly mode) "CLPROTO WILLING_TO_FORM hostname",
#   repeating every 2 seconds.
#   At same time node listens on the same port, for other CLPROTO
#   requests;
#
#     WILLING_TO_FORM [apps] - adds hostname of list of communicating nodes
#
#   When listening process has had responses from all nodes, it
#   now waits for FOLLOW_ME message from the master node (first 
#   node alphabetically). It still sends on the WILLING_TO_FORM
#   requests from a child at the same time.
#
#   Apps can be a list of failed applications which the new master
#   node needs to restart once the cluster has been formed.
#
#     FOLLOW_ME node
#
#   When this request is recieved the child process sending out the
#   "WILLING_TO_FORM" is killed as the node sets "node" as the master.
#   It also sends back a message directly to the node - 'FOLLOWER node'.
#   The response will be the status report from that node, including
#   list of available nodes, currently running applications.
#
#      What if?
#      --------
#      Gets a FOLLOW_ME before getting all responses? 
#      Allowed to join existing cluster - yes? handle FOLLOW_ME
#      otherwise abort indicating cluster already formed.
#
#      Does not get all WILLING_TO_FORM messages for all nodes in
#      timout period?
#      If the --force option is not specified then abort with error
#      indicating not all nodes responded in timeout period.
#      If force then wait for FOLLOW_ME node and treat that node as
#      master.
#
#
# WHEN A NODE BECOMES MASTER
# ==========================
# A will become a master when it is the first alphabetically in the
# list of node responses.
# If not all nodes available and no --force option then will abort.
# Otherwise will become 'master', and do the following:
# Send out "FOLLOW_ME hostname" broadcasts, and listen for 'FOLLOWER node'
# messages and handle as appropriate.
#
# When a FOLLOWER node message is received that node is added to list
# of nodes in the cluster and it will response with list of current nodes
# and current applications directly to that node.
#
# Every 5 seconds once a node becomes the master it will broadcast a 
# STATUS message indicating what it believes to be the current list of
# nodes in the cluster, and currently running applications.
#
# WHEN THE MASTER NODE DIES
# =========================
# When a node notices that the master daemon has died it will
# send out the following message:
# 
# REFORM [apps] [dead=nodes]
#
# Whenever a node gets this message it re-enters cluster formation
# node and sends out WILLING_TO_FORM etc as described above.
# 
# Once formation is complete the apps that were live on the node
# that is no longer present will be handled. It does this by
# looking at the application status table for any applications
# on nodes that are no longer running. The master then 
# handles starting them up.
# 
# SPECIAL SCENARIO: MASTER DAEMON DIES, BUT NOT NODE
# ==================================================
# In this instance the cluster is not reformed, since
# the heartbeat processes are still running. Hence when the
# cluster daemon on what is the master node sends out WILLING_TO_FORM
# no other nodes will answer it ... though in this case those nodes
# will know that node as the master, and will in turn queue a request 
# of high priority 'YOU_ARE_MASTER'. When it recieves such a request
# it knows itself as a master, and moves to sending out FOLLOW_ME to
# ascertain the nodes currently in the cluster...
#
# OTHER CLUSTER DAEMON MESSAGES HANDLED HERE
# ==========================================
#
# GET_MODDED_SETTINGS REQUESTOR=N
#		This is sent out by a cluster daemon when it joins
#		the cluster initially. It is sent to the master 
#		which will respond with [asynchronously]:
# SET_MODDED_SETTINGS DATA=YYY
#		When a node receives this it clears down any 
#		settings it currently has for the modded_settings
#		table and loads in these details.
#		YYY is a hex string - which when unpacked will be in the
#		format of "app|attribute|value" - new line separated.
#
# REG APP=app [NODES=n,...] [PROBE=0|1]
# APP_ADD=app [NODES=n,...] [PROBE=0|1]
#		Usually sent to the master when the specified app
#		is validated [and so can be registered].
#		It will then take responsibility by sending out
#		this request to the other running nodes.
#		If probe is present teh host [including master] will
#		grab an IP [if any] from the application and check if
#		defined locally. If so triggers a APP_UPDATE...
# DEREG APP=app
# APP_REMOVE APP=app
#		Used to remove the application from the configuration.
#		It will remove the status information for the application
#		and refresh local daemons. If this is the master will send
#		out messages to the other cluster daemons to do the same.
#
# 		The lha_destroyapp will have already removed references
#		to the application from the configuration databases currently
#		available.
#
# APP_UPDATE APP=app STATUS=status NODE=node [UPDATETIME=NNN] [VALID=nodes]
#		An update to application status - it is sent from
#		a node when the application starts or stops.
#		[Sent to master to distribute if necessary]
# APP_CHNODES	Change the list of nodes that are considered valid
#		for an application. A node can nominate itself as
#		invalid/valid [due to software failures] or if a
#		node is specified/validated as suitable for
#		hosting the application in question.
#		[Sent to master to distribute if necessary]
# VALID_NODES	NODES node,...
#		Sends a list of nodes to running nodes indicating
#		which nodes are currently in the cluster.
# NODE_FAILURE  NODES n[,..] [ELECTIVE 0|1] [SOURCE host]
#		Sent a comma separated list of nodes that are not available.
#		This message is sent from the heartbeat daemon on a server to
#		the local cluster daemon. This daemon will simply send it 
#		onwards to the master [obviously removing the nodes from
#		the list of valid nodes it keeps.
#		When the master gets this message [drops duplicates] it will
#		work out where any apps should be distributed to and sends
#		out requests as appropriate.
#
#		However if the failed node is thought to be the current master
#		then things are different - the cluster must enter a reformation
#		mode to select a new master.
#
#		The source is optional but is useful for debugging since it allows
# 		us to understand who is requesting the node failure handling.
#
# NODE_REMOVE	NODE n
#		Called when the specified node is no longer part of the cluster.
#		This is sent to the master to distribute to the other nodes as well
#		as action itself.
#
# NODE_ADD	NODE n
#		Called when a specified node has been added to the cluster.
#		This is sent to the master to distribute to the other nodes as well
#		as action itself. Functionally almost equivalent to NODE_REMOVE.
#
# VALID_APPS	A list of the currently registered apps 
#		available to be run.
# APP_STATUS	A | separated list where each element is:
#		appname,node,status,updatetime - so each node has information
#		where each app is running.
#		APPINFO is variable name passed
#		The above three messages are sent out from the 
#		to all [running] nodes every 10 seoncds.
#		This information is sent out by the node that is
#		currently the master.
#
# MASTER	Returns the name of the node this node thinks is the
# 		master.
#
#		A node will only send out the status information it
#		has running locally. It should also record which are
#		active and if any do become inactive it should send
#		out that message and then not again.
#
#		What happens when this node sends out an inactive
#		message, but they is after another starting message
#		is sent out from elsewhere??
#		In these scenarios the updatetime becomes important -
#		if it is older than the updatetime currently registered,
#		or is the same time, it is ignored.
#



use strict;
use cldb;
use C_client2;
use C_server2;
use clstatdb;
use clutils qw(nextnode map_select_output);
use daemon_connect;
use clevents;
use File::Basename;
use Carp qw(cluck confess);
use Sys::Hostname;
use POSIX ":sys_wait_h";
use clreq;
use clreqdxml;
use Data::Dumper;

#########################################################################
# formreqs are simple requests with a mandatory form which we do not	#
# attempt to parse.							#
#########################################################################


my @clproto_formreqs=(
	"WILLING_TO_FORM",
	"FOLLOWER",
	"FOLLOW_ME",
	"YOU_ARE_MASTER",
);

#########################################################################
# Requests in this table must confirm to specific arguments, though of	#
# course no specific order is required.					#
#########################################################################

my %clproto_requests=(
	GET_MODDED_SETTINGS => {REQUIRED => ["REQUESTOR"],
				OPTIONAL => []},
	SET_MODDED_SETTINGS => {REQUIRED => ["DATA"],
				OPTIONAL => []},
	REG		=> {	REQUIRED => ["APP"],
				OPTIONAL => ["PROBE","NODES"]},
	APP_ADD		=> {	REQUIRED => ["APP"],
				OPTIONAL => ["PROBE","NODES"]},
	UNREG 		=> {	REQUIRED => ["APP"], 
				OPTIONAL => []},
	APP_REMOVE	=> {	REQUIRED => ["APP"], 
				OPTIONAL => []},
	VALID_NODES	=> {	REQUIRED => ["NODES"],
				OPTIONAL => []},
	APP_STATUS	=> {	REQUIRED => ["APP_INFO"],
				OPTIONAL => []},
	REFORM		=> {	REQUIRED => [],
				OPTIONAL => ["DEAD"]},
	YOU_ARE_MASTER	=> {	REQUIRED => [],
				OPTIONAL => []},
	WHOIS_MASTER	=> {	REQUIRED => [],
				OPTIONAL => []},
	NODE_FAILURE	=> {	REQUIRED => ["NODES"],
				OPTIONAL => ["ELECTIVE","SOURCE"]},
	NODE_REMOVE	=> {	REQUIRED => ["NODE"],
				OPTIONAL => []},
	NODE_ADD	=> {	REQUIRED => ["NODE"],
				OPTIONAL => []},
	APP_UPDATE	=> {	REQUIRED => ["APP","STATUS","NODE"],
				OPTIONAL => ["UPDATETIME","VALID"]},
);

sub _istrue {
my $x=uc($_[0]);

	if(scalar(grep {$_ eq $x} ("TRUE",1,"YES"))) {
		return 1;
	}
	return 0;
}

sub _isfalse {
	return !_istrue($_[0]);
}

my $error_rc=0;
my $error_msg=undef;

my $DEFAULT_formtime=120;	# If not defined, and missing from
				# configuration database.
my $DEFAULT_bcast_threshold=9;	# Number of nodes after which broadcasting
				# should be used.

sub logit {
my $self=shift;
my $type=shift;
my $cline;
my $xx;

	return if !defined($self->{logger});
	for $cline (@_) {
		my @F=split(/\n/,$cline);
		for $xx (@F) {
			$self->{logger}->msg($type,$xx);
		}
	}
}

sub msg {
my $self=shift;

	logit($self,"msg",@_);
}

sub warning {
my $self=shift;

	logit($self,"warn",@_);
}

sub debug {
my $self=shift;

	logit($self,"debug",@_);
}

sub error {
my $self=shift;
my $rc;

	if(scalar(@_)>1) {
		my $x=scalar(@_)-1;
		$rc=$_[$x];
		@_=@_[0 .. $x-1];
	}
	logit($self,"error",@_);
	exit($rc) if defined($rc);
}

#########################################################################
# [logger]	Logger object for logging messages.			#
# [pki]         Set to 1 to use pki, default is 0.			#
# configdb	Name of configuration database file.			#
# statusdb	Name of status database (will update when appropriate	#
#		to ensure querying routines get accurate picture).	#
# nodes		A hash ref where each name is the name of a node, and	#
#		each value is a comma-separated list of IP addresses for#
#		that node.						#
#									#
# Will keep name of config.db and pull out details when required.	#
#########################################################################

sub new {
my $self={};
my $class=shift;
my %args=@_;

	$self->{SET_NODE_LIST}=[];
	$error_rc=0;
	$error_msg=undef;
	if(!exists($args{configdb})) {
		cluck("Missing mandatory 'configdb' argument.");
	}
	if(!exists($args{statusdb})) {
		cluck("Missing mandatory 'statusdb' argument.");
	}
	if(!exists($args{nodes})) {
		cluck("Missing mandatory 'nodes' argument.");
	}
	$self->{last_reform}=0;
	$self->{last_reform_dead_nodes}=[];
	$self->{nodes}=$args{nodes};
	if(! -f $args{configdb}) {
		cluck("Configuration database '$args{configdb}' does not exist.");
	}
	$self->{configdb}=$args{configdb};
	$self->{statusdb}=$args{statusdb};
	$self->{logger}=$args{logger} if exists($args{logger}) && defined($args{logger});
	$self->{pki}=1;
	$self->{pki}=$args{pki} if exists($args{pki}) && defined($args{pki});
	$self->{configvars_cache}={};
	$self->{statusvars_cache}={};
	bless($self,$class);
	my $clustername=_getconfigvar($self,"clustername");
	if(!defined($clustername)) {
		$clustername="?unknown?";
	}
	$self->{clevents}=new clevents($clustername);
	debug($self,"New clprotocol object created.");
	return $self;
}

#########################################################################
# When RECONFIGURE is received by the cluster daemon it will eventually	#
# call this routine to clear out the cached settings forcing them in	#
# from the database again.						#
#########################################################################

sub reconfigure($) {
my $self=$_[0];

	$self->{SERVER_SOCKET}->load_window_keys();
	$self->{configvars_cache}={};
	$self->{statusvars_cache}={};
}

sub _getconfigvar {
my $self=shift;
my  $var=shift;

	$error_rc=0;
	$error_msg=undef;

	#################################################################
	# If value is cached do not bother going to database, 		#
	# otherwise get all vars into the cache this time round.	#
	#################################################################
	if(exists($self->{configvars_cache}->{$var})) {
		return $self->{configvars_cache}->{$var};
	}
	my $db=new cldb($self->{configdb});
	if(!defined($db)) {
		($error_rc,$error_msg)=cldb::error;
		return undef;
	}
	my $hlist=$db->get_rows("SELECT var,value FROM globals");
	$db->disconnect;
	if(!defined($hlist)) {
		($error_rc,$error_msg)=cldb::error;
		return undef;
	}
	for (@$hlist) {
		debug($self,"clprotocol: _getconfigvar: loaded $_->{var} == $_->{value}");
		$self->{configvars_cache}->{$_->{var}}=$_->{value};
	}
	if(exists($self->{configvars_cache}->{$var})) {
		return $self->{configvars_cache}->{$var};
	}
	return undef;
}

sub _setconfigvar {
my $self=shift;
my ($var,$value)=@_;

	cluck("Missing mandatory 'var' setting.") if !defined($var);
	$value="" if !defined($var);
	$self->{configvars_cache}->{$var}=$value;
	my $db=new cldb($self->{configdb});
	if(!defined($db)) {
		($error_rc,$error_msg)=cldb::error;
		return 0;
	}
	if(! $db->do("INSERT OR REPLACE INTO globals (var,value) VALUES ('$var','$value');")) {
		$db->disconnect;
		($error_rc,$error_msg)=cldb::error;
		return 0;
	}
	$db->disconnect;
	return 1;
}

sub _getstatusvar {
my $self=shift;
my  $var=shift;

	$error_rc=0;
	$error_msg=undef;

	#################################################################
	# If value is cached do not bother going to database, 		#
	# otherwise get all vars into the cache this time round.	#
	#################################################################
	if(exists($self->{statusvars_cache}->{$var})) {
		return $self->{statusvars_cache}->{$var};
	}

	my $db=new clstatdb($self->{statusdb});
	if(!defined($db)) {
		($error_rc,$error_msg)=clstatdb::error;
		return undef;
	}

	#################################################################
	# The table might not exist, so create it now...		#
	#################################################################

	if(! $db->exists("table","cluster_status_vars")) {
		if(! $db->do("CREATE TABLE cluster_status_vars(var VARCHAR(50) PRIMARY KEY, value VARCHAR(200) );")) {
			$db->disconnect;
			($error_rc,$error_msg)=clstatdb::error;
			return undef;
		}
	}

	my $hlist=$db->get_rows("SELECT var,value FROM cluster_status_vars");
	$db->disconnect;
	if(!defined($hlist)) {
		($error_rc,$error_msg)=cldb::error;
		return undef;
	}
	for (@$hlist) {
		$self->{statusvars_cache}->{$_->{var}}=$_->{value};
	}
	if(exists($self->{statusvars_cache}->{$var})) {
		return $self->{statusvars_cache}->{$var};
	}
	return undef;
}

sub _setstatusvar {
my $self=shift;
my ($var,$value)=@_;
my $href;

	cluck("Missing mandatory 'var' setting.") if !defined($var);
	$value="" if !defined($var);
	if(ref($value)) {
		$href=$value;
	}

	if($var =~ /:clusternodestate$/) {
		my ($server,$j)=split(/:/,$var);
		msg($self,"Set node_state->$server to $value");
		$self->{node_state}->{$server}=$value;
	}
	#################################################################
	# Optimisation - if the value that is being set is the same as	#
	# the existing value then exit - saving open/close of DB.	#
	#################################################################

	if(!defined($href)) {
		msg($self,"var=$var,value=$value");
		if(exists($self->{statusvars_cache}->{$var}) && defined($self->{statusvars_cache}->{$var}) && $value eq $self->{statusvars_cache}->{$var}) {
			return 1;
		}
		$self->{statusvars_cache}->{$var}=$value;
	} else {
		my $hkey;
		for $hkey (sort(keys(%$href))) {
			$self->{statusvars_cache}->{"$hkey:$var"}=$href->{$hkey};
			msg($self,"Set $hkey:$var to $href->{$hkey}");
		}
	}

	#################################################################
	# Attempt to open database / create table if necessary.		#
	#################################################################

	my $db=new clstatdb($self->{statusdb});
	if(!defined($db)) {
		($error_rc,$error_msg)=clstatdb::error;
		return 0;
	}
	if(! $db->create_if_needed("cluster_status_vars")) {
		$db->disconnect;
		cluck("Unable to create 'cluster_status_vars' table!");
		return 0;
	}
	if(!defined($href)) {
		if(! $db->do("INSERT OR REPLACE INTO cluster_status_vars (var,value) VALUES ('$var','$value');")) {
			$db->disconnect;
			($error_rc,$error_msg)=clstatdb::error;
			return 0;
		}
	} else {
		my $hkey;
		for $hkey (keys(%$href)) {
			my $value=$href->{$hkey};
			if(! $db->do("INSERT OR REPLACE INTO cluster_status_vars (var,value) VALUES ('$hkey:$var','$value');")) {
				$db->disconnect;
				($error_rc,$error_msg)=clstatdb::error;
				return 0;
			}
		}
	}
	$db->disconnect;
	$db=undef;
	return 1;
}

# Form the cluster, with [optional] settings:
#
# pki		- Use PKI or key/window protocol
# [formtime]	- Maximum amount of time to wait for teh cluster to
#		  form. If not specified will take the global variable
#		  'formtime'.
# [force]	- If set to 1 will still continue attempting to form
#                 the cluster even if not all responses recieved from
#                 all nodes in the timeout period.
# [join]	- Set to 1 if allow it to join an existing running
#		  cluster.
# [bcast_threshold]
#		- Node count after which broadcasting will be used.
#                 Takes from globals if not set, and default is 9.
# [reqnodes]	- The nodes necessary to form a cluster - if others
#		  join them it is up to them!! However once this
#		  number of nodes have returned responses cluster
#		  formation completes.
# [failed_apps]	- a comma-separated list of apps that need to be 
#		  restarted since this is a reformation of the cluster
#		  due to a failed master node.

sub form_cluster {
my $self=shift;
my %args=@_;
my $formtime=-1;
my $join=0;
my $force=0;
my $bcast_threshold=-1;
my @req_nodes=();
my $failed_apps="";

	#################################################################
	# Req nodes is set if just a subset of nodes have 		#
	# been requested.						#
	#################################################################

	if(!exists($args{pki})) {
		confess("Missing mandatory pki argument.");
	}
	$failed_apps=$args{failed_apps} if (exists($args{failed_apps}) && 
		defined($args{failed_apps})); 
	$self->{failed_apps}=$failed_apps;

	@req_nodes=split(/,/,$args{reqnodes}) if exists($args{reqnodes});

	if(exists($args{formtime})) {
		$formtime=$args{formtime};
		msg($self,"clprotocol: 'formtime' set to '$formtime' (from args).");
	}
	if($formtime==-1) {
		my $x=_getconfigvar($self,"formtime");
		if(defined($x)) {
			$formtime=int($x);
			msg($self,"clprotocol: 'formtime' set to '$formtime' (from configdb).");
		}
	}
	if($formtime==-1) {
		$formtime=$DEFAULT_formtime;
		msg($self,"clprotocol: 'formtime' set to '$formtime' (inbuilt default).");
	}
	$self->{formtime}=$formtime;

	if(exists($args{bcast_threshold})) {
		$bcast_threshold=$args{bcast_threshold};
		msg($self,"clprotocol: 'bcast_threshold' set to '$bcast_threshold' (from args).");
	}
	if($bcast_threshold==-1) {
		my $x=_getconfigvar($self,"bcast_threshold");
		if(defined($x)) {
			$bcast_threshold=int($x);
			msg($self,"clprotocol: 'bcast_threshold' set to '$bcast_threshold' (from configdb).");
		}
	}
	if($bcast_threshold==-1) {
		$formtime=$DEFAULT_bcast_threshold;
		msg($self,"clprotocol: 'bcast_threshold' set to '$bcast_threshold' (inbuilt default).");
	}


	if($args{force}) { 
		$force=$args{force};
		msg($self,"clprotocol: 'force' option set to '$force' (from args).");
	} else {
		msg($self,"clprotocol: 'force' option set to '$force' (inbuilt default).");
	}
	if($args{join}) {
		$join=$args{join};
		msg($self,"clprotocol: 'join' option set to '$join' (from args).");
	} else {
		msg($self,"clprotocol: 'join' option set to '$join' (inbuilt default).");
	}

	my $x=_getconfigvar($self,"warntime");
	if(defined($x)) {
		msg($self,"clprotocol: 'warntime' option set to '$x' (from configdb).");
	} else {
		msg($self,"clprotocol: 'warntime' option set to '10' (inbuilt default).");
	}
	$x=_getconfigvar($self,"failtime");
	if(defined($x)) {
		msg($self,"clprotocol: 'failtime' option set to '$x' (from configdb).");
	} else {
		msg($self,"clprotocol: 'failtime' option set to '20' (inbuilt default).");
	}
	if(@req_nodes) {
		msg($self,"clprotocol: Required nodes: $args{reqnodes}");
	} else {
		msg($self,"clprotocol: Required node list not specified - expecting all nodes to join cluster.");
	}
	$self->{bcast_threshold}=$bcast_threshold;
	$self->{join}=$join;
	$self->{force}=$force;
	$self->{reqnodes}=[@req_nodes];

	#################################################################
	# Before any real processing can take place we need to get the	#
	# port to listen on (parent) - and potentially broadcast on	#
	# child.							#
	#################################################################

	my $db=new cldb($self->{configdb});
	if(!defined($db)) {
		($error_rc,$error_msg)=cldb::error;
		cluck("clprotocol: Unable to open database: $error_msg");
	}

	my $href=$db->get_rows("SELECT number FROM ports WHERE detail = 'lha_clusterd';");
	if(!defined($href) || !defined($href->[0])) {
		error($self,"clprotocol: Port not found for 'lha_clusterd' in config database.",1);
	}

	my $cluster_port=$href->[0]->{number};
	$self->{cluster_port}=$cluster_port;
	error($self,"clprotocol: Unable to ascertain port to listen on from config database!",1) if !defined($cluster_port);
	msg($self,"clprotocol: Port to listen on is '$cluster_port'.");

	#################################################################
	# The next step is to load the details of all hosts/IP 		#
	# addresses, and work out whether to broadcast or direct	#
	# requests to individual hosts.					#
	#################################################################

	$db->disconnect;
	my $nc=scalar(keys(%{$self->{nodes}}));
	if($nc<$bcast_threshold) {
		msg($self,"clprotocol: Node count is '$nc' - direct IP connections used when possible.");
		$self->{broadcast}=0;
	} else {
		msg($self,"clprotocol: Node count is '$nc' - broadcasting used when possible.");
		$self->{broadcast}=1;
	}
	my $PKI=$self->{pki};
	my %NODES2=();
	if($self->{broadcast}==0) {
		my $chost;
		my $count=0;
		for $chost (sort(keys(%{$self->{nodes}}))) {
			my @ips=split(/,/,$self->{nodes}->{$chost});
			$NODES2{$chost}={};
			my $cip;
			for $cip (@ips) {
				my $channel=new C_client2(PKI=>$PKI,SERVER_INET=>$cip,SERVER_PORT=>$cluster_port,SERVER_NAME=>$chost);
				if(!defined($channel)) {
					warning($self,"clprotocol: Unable to communicate to host '$chost' via IP '$cip'.");
					warning($self,"clprotocol: Reason : " . C_client2::error_str() );
				} else {
					$channel->set_params(TIMEOUT=>1);
					$NODES2{$chost}->{$cip}=$channel;
					$count++;
				}
			}
		}
		msg($self,"clprotocol: Allocated $count channels for cluster formation messages.");
		$self->{channels}=\%NODES2;
		# channels is host->ip = channel
	} else {
		my %CHANNELS=();
		my $count=0;
		my $chost;
		for $chost (sort(keys(%{$self->{nodes}}))) {
			my @ips=split(/,/,$self->{nodes}->{$chost});
			my $cip;
			for $cip (@ips) {
				my @I=split(/\./,$cip);
				my $broadcast_addr="$I[0].$I[1].$I[2].255";
				if(!exists($CHANNELS{$broadcast_addr})) {
					my $channel=new C_client2(PKI=>$PKI,SERVER_INET=>$broadcast_addr,SERVER_PORT=>$cluster_port,SERVER_NAME=>"broadcast");
					if(!defined($channel)) {
						warning($self,"clprotocol: Unable to communicate to broadcast address '$broadcast_addr'.");
						warning($self,"clprotocol: Reason : " . C_client2::error_str() );
					} else {
						$CHANNELS{$broadcast_addr}=$channel;
						$count++;
					}
				}
			}
		}
		msg($self,"clprotocol: Allocated $count broadcast channels for cluster formation messages.");
		$self->{channels}=\%CHANNELS;
		# channels is broadcast addr
	}

	#################################################################
	# At this point the process in question now spawns a child	#
	# process that sends out the "CLPROTO WILLING_TO_FORM host" 	#
	# until further notice. This request is sent across all 	#
	# channels in send-only mode...					#
	#################################################################

	my $child_pid=fork();
	if(!defined($child_pid)) {
		error($self,"clprotocol: Failed to fork process: $!",1);
	}
	$self->{child_pid}=$child_pid;
	my $res;
	if(! $child_pid) {
		_child_cluster_form($self);
		# The above will exit when complete.
	} else {
		$res=_parent_cluster_form($self);
	}
	#################################################################
	# We we get back here the child_pid should have gone away. If	#
	# not we abort with error.					#
	#################################################################
	kill 1,$child_pid;
	#################################################################
	# return $res = 0 - failed, 1 - ok.				#
	#################################################################
	
	return $res;
}

sub _child_cluster_form {
my $self=shift;
my $hostname=hostname;
my $failed_apps=$self->{failed_apps};

	#################################################################
	# Broadcast on all channels here, until told to die!		#
	#################################################################
	my $broadcast=$self->{broadcast};
	my $channels=$self->{channels};
	$clprotocol::KEEP_ON_RUNNING=1;

	$::SIG{INT} = sub { $clprotocol::KEEP_ON_RUNNING=0; };
	$::SIG{HUP} = sub { $clprotocol::KEEP_ON_RUNNING=0; };
	while($clprotocol::KEEP_ON_RUNNING) {
		my $chan;

		#########################################################
		# We keep sending out the messages on all channels	#
		# every second until told otherwise!			#
		#########################################################

		for $chan (keys(%$channels)) {
			last if !$clprotocol::KEEP_ON_RUNNING;
			if($broadcast==0) {
				my ($cnode,$cip);
				for $cnode (keys(%$channels)) {
					last if !$clprotocol::KEEP_ON_RUNNING;
					for $cip (keys(%{$channels->{$cnode}})) {
						msg($self,"clprotocol: sending WILLING_TO_FORM -> $cnode/$cip");
						$channels->{$cnode}->{$cip}->send_request("CLPROTO WILLING_TO_FORM $hostname $failed_apps",1);
					}
				}
			} else {
		# MORE HERE
				;
			}

		}
		select(undef,undef,undef,1);
	}
	if($broadcast==0) {
		my ($cnode,$cip);
		for $cnode (keys(%$channels)) {
			for $cip (keys(%{$channels->{$cnode}})) {
				undef $channels->{$cnode}->{$cip};
			}
		}
	}
	exit(0);
	;
}

sub _clean_channels {
my $self=shift; 
my $broadcast=$self->{broadcast};
my $channels=$self->{channels};

	if(! $broadcast) {
		my ($chost,$cip);
		for $chost (keys(%$channels)) {
			for $cip (keys(%{$channels->{$chost}})) {
				undef $channels->{$chost}->{$cip};
				delete $channels->{$chost}->{$cip};
			}
			delete $channels->{$chost};
		}
	} else {
		my ($cbroad);
		for $cbroad (keys(%$channels)) {
			undef $channels->{$cbroad};
			delete $channels->{$cbroad};
		}
	}
	delete $self->{channels};
}

#########################################################################
# Return a hash of node names and node states.				#
#########################################################################

sub get_node_states {
my $self=shift;

	my %f=(%{$self->{node_state}});
	return \%f;
}

sub _parent_cluster_form {
my $self=shift;
my $pki=$self->{pki};
my $cluster_port=$self->{cluster_port};
my $server;
my $hostname=hostname;
my $broadcast=$self->{broadcast};
my @req_nodes;

	@req_nodes=@{$self->{reqnodes}};
	#################################################################
	# Since this might be called on a cluster reformation it might	#
	# already exist, so only create it if necessary.		#
	#################################################################
	if(!exists($self->{SERVER_SOCKET}) || !defined($self->{SERVER_SOCKET})) {
		$server=new C_server2(PKI => $pki,PORT => $cluster_port);
		$self->{SERVER_SOCKET}=$server;
	} else {
		$server=$self->{SERVER_SOCKET};
	}

	#################################################################
	# Start listener on port and handle messages as appropriate...	#
	#################################################################
	$clprotocol::KEEP_ON_RUNNING=1;

	$::SIG{INT} = sub { $clprotocol::KEEP_ON_RUNNING=0; };
	$::SIG{HUP} = sub { $clprotocol::KEEP_ON_RUNNING=0; };

	#################################################################
	# Remeber to write out status changes every so oftern as well!	#
	#################################################################

	my $start_time=time;
	my $last_time=$start_time+$self->{formtime};
	my $node_response_count=0;
	my $wanted_response_count=scalar(keys(%{$self->{nodes}}));
	if(@req_nodes>0) {
		$wanted_response_count=scalar(@req_nodes);
		msg($self,"clprotocol: Require $wanted_response_count nodes to form cluster.");
	} else {
		msg($self,"clprotocol: Expect all nodes [$wanted_response_count] to form/join cluster.");
	}

	my %node_responses=();
	my %node_failed_apps=();
	$node_responses{$_}="No_response" foreach (keys(%{$self->{nodes}}));
	$node_failed_apps{$_}="" foreach (keys(%{$self->{nodes}}));

	cluck("Unable to update status database!") if ! _setstatusvar($self,"clusternodestate",\%node_responses);
	cluck("Unable to update status database!") if ! _setstatusvar($self,"master","Not_defined");
	my $master;
	while(time<$last_time && $clprotocol::KEEP_ON_RUNNING) {
		my $__time=time;
		my ($request,$err);

		($request,$err)=$server->accept(1);
		if(!defined($request)) {
			next;
		}
		my @F=split(/\s+/,$request);
		if($F[0] eq "PING") {
			$server->send_response("OK");
			next;
		}
		if($F[0] eq "CLPROTO") {
			#################################################
			# If we are forming / reforming already then	#
			# ignore this request.				#
			#################################################
			if($F[1] eq "REFORM") {
				next;
			}
			if($F[1] eq "WILLING_TO_FORM") {
				$F[3]="" if !defined($F[3]);
				if(!exists($node_responses{$F[2]})) {
					warning($self,"clprotocol: WILLING_TO_FORM from $F[2] - node not in local cluster definition - ignoring message!");
				} else {
					if($node_responses{$F[2]} eq "No_response") {
						$node_responses{$F[2]}="Willing";
						$node_failed_apps{$F[2]}=$F[3];
						msg($self,"clprotocol: Node '$F[2]' is WILLING_TO_FORM.");
						$node_response_count++;
					}
				}
			}
			#################################################
			# Whoops! I am the master but I must have	#
			# restarted - so we check to see if the heart	#
			# beat deamon is running, and if so work out	#
			# the currently running nodes based on what	#
			# it thinks.					#
			#################################################
			if($F[1] eq "YOU_ARE_MASTER") {
				msg($self,"clprotocol: Node '$F[2]' sent YOU_ARE_MASTER!");
				my $db=new clstatdb($self->{statusdb});
				if(!defined($db)) {
					my ($error_rc,$error_msg)=clstatdb::error;
					cluck("Unable to open status database: $error_msg");
					$server->no_response;
					next;
				}
				# Query the node_status table...	#
				my $max=_getconfigvar($self,"failtime");
				$self->{node_state}->{$_}="No_response" foreach (keys(%{$self->{nodes}}));
				my $hlist=$db->get_rows("SELECT node,update_time FROM node_status;");
				$db->disconnect; $db=undef;
				if(!defined($hlist)) {
					my ($error_rc,$error_msg)=cldb::error;
					cluck("Unable to get node_status details: $error_msg");
					$server->no_response;
					next;
				}
				my $t=time;
				for (@$hlist) {
					if($_->{update_time}+$max>=$t) {
						$self->{node_state}->{$_->{node}}="Joined";
					}
				}
				my $h=hostname;
				$self->{node_state}->{$h}="Joined";
				cluck("Unable to update status database!") if ! _setstatusvar($self,"clusternodestate",$self->{node_state});
				$self->{MASTER}=hostname;
				$server->no_response;
				kill 1,$self->{child_pid}; #REQUIRED??
				_clean_channels($self);
				cluck("Unable to update status database!") if ! _setstatusvar($self,"master",hostname);
				return 1;
			}
			#################################################
			# If a master node has all responses it might 	#
			# send out this request before this server	#
			#################################################

			if($F[1] eq "FOLLOW_ME") {
				my ($resp,$err);
				$master=$F[2];
				msg($self,"clprotocol: Got FOLLOW_ME from '$master' - will attempt to join.");
				cluck("Unable to update status database!") if ! _setstatusvar($self,"master",$master);
				cluck("Unable to update status database!") if ! _setstatusvar($self,"$master:clusternodestate","Joined");
				$self->{MASTER}=$master;
				msg($self,"clprotocol: Killing child process $self->{child_pid}");
				kill 1,$self->{child_pid};
				#################################
				# Remove all channels that 	#
				# might be open for writing,	#
				# leaving just server socket.	#
				#################################
				debug($self,"clprotocol: queued request: CLPROTO FOLLOWER $hostname");
				queue_request($self,REQUEST=>"CLPROTO FOLLOWER $hostname",
							TO => $master, PRIORITY => 5);
				my $ccc=process_queue($self,99);
				{
					my $db=new clstatdb($self->{statusdb});
					if(defined($db)) {
						$db->do("DELETE FROM modded_settings;");
					}
					$db->disconnect();
				}

				_clean_channels($self);
				return 1;
			}
		} else {
			msg($self,"clprotocol: Unexpected message[1] - ignoring '$request'.");
		}
		$server->no_response;

		#########################################################
		# If we have specifically asked to join a cluster then	#
		# we ignore the bits below since we are waiting for the	#
		# current master to send us something...
		#########################################################

		if($self->{join}) {
			msg($self,"clprotocol: No response from master - still waiting...");
			next;
		}

		#########################################################
		# Jump out of the loop if the required number of	#
		# nodes via req_nodes have responded.			#
		#########################################################
		if(@req_nodes) {
			my $cnode;
			my $abort_loop=1;
			for $cnode (@req_nodes) {
				if(!exists($node_responses{$cnode})) {
					$abort_loop=0;
					last;
				}
				if($node_responses{$cnode} ne "Willing") {
					$abort_loop=0;
					last;
				}
			}
			if($abort_loop) {
				msg($self,"clprotocol: All required nodes responed - continuing cluster formation.");
				last;
			}
		}

		#########################################################
		# We jump out of the loop if all nodes respond.		#
		#########################################################
		last if $node_response_count==$wanted_response_count;

	}
	if(! $clprotocol::KEEP_ON_RUNNING) {
		warning($self,"Aborted cluster formation on signal.");
		exit(1);
	}

	#################################################################
	# Got here, so check node_response_count against 		#
	# wanted_response_count and if differ then abort/join/form	#
	# depending on other vars!!					#
	#################################################################

	if($node_response_count != $wanted_response_count) {
		msg($self,"clprotocol: Handle timeout event ($node_response_count nodes responded).");

		#########################################################
		# if $self->{force}==0 then abort.			#
		#########################################################

		if($self->{force}==0) {
			error($self,"Only $node_response_count of $wanted_response_count nodes responsed in time out window. Aborting.");
			print STDERR "Error: Only $node_response_count of $wanted_response_count nodes responsed in time out window. Aborting.\n";
			exit(1);
		}
		warning($self,"Only $node_response_count of $wanted_response_count nodes responsed in time out window. Continuing cluster form without all nodes.");
	}

	#################################################################
	# At least some nodes (might just be 1 - myself) are willing	#
	# to form a cluster, so work out which.				#
	#################################################################
	for (sort(keys(%{$self->{nodes}}))) {
		if($node_responses{$_} eq "Willing") {
			$master=$_; last;
		}
	}
	if(!defined($master)) {
		error($self,"Not even got a message from myself! Is IP address of local machine valid!!?? Aborting.");
		kill 1,$self->{child_pid};
		exit(1);
	}

	cluck("Unable to update status database!") if ! _setstatusvar($self,"clusternodestate",\%node_responses);
	cluck("Unable to update status database!") if ! _setstatusvar($self,"master",$master);
	$self->{MASTER}=$hostname;
	if($master eq $hostname) {
		msg($self,"clprotocol: Expect master to be '$master' ... thats me!");
		cluck("Unable to update status database!") if ! _setstatusvar($self,"$hostname:clusternodestate","Joined");
		kill 1,$self->{child_pid};
	} else {
		msg($self,"clprotocol: Expect master to be '$master'.");
	}

	#################################################################
	# If current node is master than for each other node send out	#
	# requests to join (and discard them from cluster if they do	#
	# not response) - otherwise add them to list of joined nodes.	#
	#################################################################

	$self->{node_state}=\%node_responses;
	if($master eq $hostname) {
		#########################################################
		# Clear down the status database if this is the		#
		# master node.						#
		#########################################################

		#########################################################
		# We also look at all the entries we have in the	#
		# node_failed_apps() listing and generate a list of 	#
		# apps that we need to look at starting.		#
		#########################################################

		my %fapps=();
		for my $ckey (keys %node_failed_apps) {
			my @F=split(/,/,$node_failed_apps{$ckey});
			$fapps{$_}=1 foreach(@F);
		}

		{
			#################################################
			# Recort the information in restart_apps - 	#
			# THOUGH THIS IS NOT YET USED AS PART OF THE	#
			# CLUSTER REFORMATION PROCESS.			#
			#################################################
			my $db=new clstatdb($self->{statusdb});
			if(defined($db)) {
				$db->create_if_needed("restart_apps");
				my $t=time;
				$db->do("DELETE FROM modded_settings;");
				for my $ckey (keys %fapps) {
					$db->do("DELETE FROM restart_apps WHERE appname = '$ckey';");
					$db->do("INSERT INTO restart_apps (appname,failnode,failtime) VALUES('$ckey','?','$t'");
				}
			}
			$db->disconnect();
		}
		$node_responses{$master}="Joined";
		
		cluck("Unable to update status database!") if ! _setstatusvar($self,"clusternodestate",\%node_responses);
		#########################################################
		# 2nd stage cluster formation, might take some time	#
		# so define it as form_time again...			#
		#########################################################

		my $start_time=time;
		my $last_time=$start_time+$self->{formtime};
		my @OL=sort(keys(%node_responses));
		while(time<=$last_time) {
			my ($cip,$add_to_cluster);
			my ($resp,$err);

			$add_to_cluster=undef;
			for (@OL) {
				if($node_responses{$_} eq "Willing") {
					$add_to_cluster=$_; last;
				}
			}
			last if !defined($add_to_cluster);
			msg($self,"clprotocol: Sending FOLLOW_ME to '$add_to_cluster'.");

			#################################################
			# Ask node about joining cluster ...		#
			# but always assume they will unless told	#
			# otherwise later in mainloop.			#
			#################################################

			my $joined=0;
			if(! $broadcast) {
				for $cip (keys(%{$self->{channels}->{$add_to_cluster}})) {
					($resp,$err)=$self->{channels}->{$add_to_cluster}->{$cip}->send_request("CLPROTO FOLLOW_ME $hostname",1);
					msg($self,"Sent: CLPROTO FOLLOW_ME $hostname to $add_to_cluster/$cip");
					if(defined($resp)) {
						$joined=1; last;
					}
				}
			} else {
				# MORE HERE
				;
			}

			#################################################
			# Update ndoe_responses and status database if	#
			# did join cluster.				#
			#################################################
			if($joined) {
				$node_responses{$add_to_cluster}="Joined";
				cluck("Unable to update status database!") if ! _setstatusvar($self,"$add_to_cluster:clusternodestate","Joined");
			}
		}
		#################################################
		# Gobble up any waiting requests..., do		#
		# nothing with them until the cluster has	#
		# been formed correctly - returns a RETRY	#
		# indicating the client should retry soon.	#
		#################################################
		select(undef,undef,undef,.25);
		while(1) {
			my ($request,$err)=$server->accept(.25);
			last if !defined($request);
			msg($self,"Master gobbled: $request");
			($request,$err)=$server->send_response("CLPROTO RETRY $hostname");
		}

		#########################################################
		# Either timeout, or all nodes responded.		#
		#########################################################

		if(time>$last_time) {
			# Handle timeout!				#
			warning($self,"Some nodes did not join in expected window - will check periodically.");
			_clean_channels($self);
			return 1;
		}
		msg($self,"clprotocol: All nodes expected to join cluster did do so.");
		kill 1,$self->{child_pid}; #REQUIRED??
		_clean_channels($self);
		return 1;
	} else {
		#########################################################
		# If not the master we wait for a FOLLOW_ME request.	#
		#########################################################
		my $start_time=time;
		my $last_time=$start_time+$self->{formtime};
		my $joined=0;
		while(time<=$last_time) {
			my ($request,$err);

			($request,$err)=$server->accept(1);
			next if !defined($request);
			#################################################
			# Do we still need this? Certainly can't send 	#
			# a response since the socket will be closed.	#
			#################################################
			if($request =~ /^CLPROTO WILLING_TO_FORM/) {
				#########################################
				# Ignore willing_to_form since we are	#
				# not the master node.			#
				#########################################
				next;
			}
			if($request =~ /^CLPROTO RETRY/) {
				$server->send_response("CLPROTO OK $hostname");
			}
			if($request =~ /^CLPROTO FOLLOW_ME/) {
				my @F=split(/\s+/,$request);

				if($F[2] ne $master) {
					msg($self,"Got FOLLOW_ME from $F[2] - not $master!");
					$master=$F[2];
				} else {
					msg($self,"Got FOLLOW_ME from expected master $master.");
				}
					
				$self->{MASTER}=$master;
				queue_request($self,REQUEST=>"CLPROTO FOLLOWER $hostname",
							TO => $master, PRIORITY => 5);
				cluck("Unable to update status database!") if ! _setstatusvar($self,"$hostname:clusternodestate","Joined");
				cluck("Unable to update status database!") if ! _setstatusvar($self,"master",$master);
				$joined=1;
				last;
			} else {
				warning($self,"clprotocol: Unexpected message[2] - ignoring '$request'.");
			}
		}
		if(! $joined) {
			error($self,"Failed to join forming cluster during 2nd stage formation.");
			print STDERR "Failed to join forming cluster during 2nd stage formation.\n";
			exit 1;
		}
		kill 1,$self->{child_pid};
	}
	_clean_channels($self);
	return 1;
}

#########################################################################
# The below is called only on the master - only at cluster start-up.	#
# It loads in the details of each APP that is validated. It will also	#
# queue a CHECK_RUNNING for each app that will check to see if the	#
# IP address is configured on the host.					#
#########################################################################

sub probe_n_register_apps {
my $self=shift;
my ($error_rc,$error_msg);

	my $db=new cldb($self->{configdb});
	if(!defined($db)) {
		($error_rc,$error_msg)=cldb::error;
		return (1,$error_msg);
	}
	my $hlist=$db->get_rows("SELECT name,nodes FROM applications WHERE validated = 1 OR validated = 'Y';");
	if(!defined($hlist)) {
		($error_rc,$error_msg)=cldb::error;
		return (1,$error_msg);
	}
	#################################################################
	# Firstly queue request to distribute that the app has been	#
	# registered.							#
	#################################################################
	for (@$hlist) {
		queue_request($self,REQUEST=>"CLPROTO REG APP=$_->{name} NODES=$_->{nodes}",
					TO => hostname, PRIORITY => 1);
	}
	my @up=();
	for (keys(%{$self->{node_state}})) {
		if($self->{node_state}->{$_} eq "Joined") {
			push @up,$_;
		}
	}
	msg($self,"up=" . join(",",@up));
	my $clreqdxml=new clreqdxml("$::ENV{TRUECL_CFG}/clreqd.xml");
	my $reqd=new clreq(hosts => [@up],clreqdxml => $clreqdxml, logger=>$::LOGGER, timeout=>6);
	my %apps=();
	msg($self,"");
	msg($self,"* * * APPLICATION PROBING : START * * *");
	msg($self,"");
	my @hl2=();
	push @hl2,$_->{name} foreach(@$hlist);
	msg($self,"Checking for applications running on the following nodes: " . join(",",@up));
	msg($self,"Checking for the following list of applications: " . join(",",@hl2));
	if(defined($reqd)) {
		for (@$hlist) {
			my $r=$reqd->run(cmd => ["probe_app","APPLICATION",$_->{name}]);
			if(!defined($r)) {
				error($self,"The 'probe_app' failed to run - aborting further probes.");
				return (0,undef);
			}
			my $n;
			for $n (keys(%$r)) {
				msg($self,"Probe of $_->{name} on $n : $r->{$n}");
				if($r->{$n} =~ /^OK UP/) {
					$apps{$_->{name}}=$n;
					last;
				}
			}
		}
	}
	my $statdb=new clstatdb($self->{statusdb});
	if(!defined($statdb)) {
		my ($error_rc,$error_msg)=clstatdb::error;
		cluck("Unable to open status database: $error_msg");
		return (0,undef);
	}
	my $ti=time;
	for (keys(%apps)) {
		msg($self,"Updating status database for application '$_'.");
		$statdb->do("INSERT OR REPLACE INTO app_status (name,host,status,updatetime) VALUES ('$_','$apps{$_}','RUNNING','$ti');");
		queue_request($self,REQUEST=>"CLPROTO APP_UPDATE APP=$_ STATUS=RUNNING NODE=$apps{$_}",
					TO => "??ALL??", PRIORITY => 2);
	}
	$statdb->disconnect();
	msg($self,"");
	msg($self,"* * * APPLICATION PROBING : COMPLETED * * *");
	msg($self,"");
	return (0,undef);
}

#########################################################################
# This routine is typically called from the main program to deal with	#
# any CLPROTO requests the daemon might get [which is probably all 	#
# for now!].								#
# It uses the clproto_requests hash to ensure the request is valid 	#
# and has been passed all mandatory arguments.				#
#########################################################################

sub deal_request {
my $self=shift;
my $req=shift;

	my %argv=();
	my $cluster_port=$self->{cluster_port};
	my $sock=$self->{SERVER_SOCKET};
	my @F=split(/\s+/,$req);
	if($F[0] ne "CLPROTO") {
		msg($self,"clprotocol: Passed non-clprotocol message ' $req'.");
		$sock->no_response;
		return 0;
	}
	my $simple=0;
	if(scalar(grep {$_ eq $F[1]} @clproto_formreqs)) {
		$simple=1;
	} else {
		if(!exists($clproto_requests{$F[1]})) {
			msg($self,"clprotocol: Passed unrecognised command '$F[1]'.");
			$sock->no_response;
			return 0;
		}

	#################################################################
	# Build argv hash, all in upper case from F2 onwards		#
	#################################################################

		{
			my $f;
			for $f (@F[2..$#F]) {
				my @z=split(/=/,$f);
				$z[1]="" if !defined($z[1]);
				$argv{uc($z[0])}=$z[1];
			}
		}

		#################################################################
		# Check all mandatory options are specified and ignore the 	#
		# request if not. Also give a warning if any unrecognised 	#
		# options are passed.						#
		#################################################################

		my $opt=$clproto_requests{$F[1]}->{OPTIONAL};
		my $man=$clproto_requests{$F[1]}->{REQUIRED};
		my ($cman,$copt);
		my @m=();
		for $cman (@$man) {
			if(!exists($argv{$cman})) {
				push @m,$cman;
			}
		}
		my @un=();
		for $copt (keys %argv) {
			if(scalar(grep {$_ eq $copt} @$man)==0 &&
				scalar(grep {$_ eq $copt} @$opt)==0) {
				push @un,$copt;
			}
		}
		if(@un) {
			warning($self,"Following unrecognised options for '$req' passed: " . join(",",@un));
		}
		if(@m) {
			error($self,"Following mandatory options missing for '$req': " . join(",",@m) . " - ignoring!");
			$sock->no_response;
			return 0;
		}
	}

	if($F[1] eq "WHOIS_MASTER") {
		if(exists($self->{MASTER}) && defined($self->{MASTER})) {
			$sock->send_response("OK $self->{MASTER}");
		} else {
			$sock->send_response("OK ??");
		}
		return 0;
	}
	#################################################################
	# If we get a REFORM then go into reform mode...		#
	#################################################################

	if($F[1] eq "REFORM") {

		$sock->no_response;
		my @x=();
		if(exists($argv{DEAD}) && length($argv{DEAD})) {
			@x=split(/,/,$argv{DEAD});
		}
		#########################################################
		# If a reform message is received less than 90 seconds	#
		# after the previous one it should be ignored for now -	#
		# assuming the dead nodes are the same!			#
		#########################################################
		if($self->{last_reform}+120>time) {
			if(exists($argv{DEAD})) {
				my $same=1;
				for my $cnode (@x) {
					if(! grep {$_ eq $cnode} @{$self->{last_reform_dead_nodes}}) {
						$same=0; last;
					}
				}
				if($same) {
					msg($self,"Reform ignored - same as previous!");
					return 0;
				}
			} else {
				msg($self,"Reform - but no dead nodes specified - will attempt.");
			}
		}
		$self->{last_reform}=time;
		$self->{last_reform_dead_nodes}=[@x];
		msg($self,"Entering reform mode[1] ... dead=$argv{DEAD}") if defined(&::msg);
		return (88,[],[@x]);
	}

	#################################################################
	# If we get a REG request and are not the master then we	#
	# update our status database. If the master we then queue	#
	# to send out to all active nodes.				#
	#################################################################

	if($F[1] eq "REG") {
		my $db=new clstatdb($self->{statusdb});
		if(!defined($db)) {
			my ($error_rc,$error_msg)=clstatdb::error();
			cluck("clprotocol: Unable to open database [$error_msg]");
			$sock->no_response;	# One way request type
			return 0;
		}
		$db->create_if_needed("app_status");
		$db->create_if_needed("app_registered");
		my $t=time;
		my $r;
		if($F[1] eq "REG") {
			$r=$db->do("INSERT OR REPLACE INTO app_registered (name,date) VALUES ('$argv{APP}','$t');");
		} else {
			$r=$db->do("DELETE FROM app_registered WHERE name = '$argv{APP}';");
		}
		if(!$r) {
			$db->disconnect;
			($error_rc,$error_msg)=clstatdb::error;
			cluck("clprotocol: Unable to create app_registered table [$error_msg]");
			$sock->no_response;
			return 0;
		}
		$sock->no_response;
		#########################################################
		# We've updated the app_registered table, we may need	#
		# to check for other arguments...			#
		#########################################################
		if($F[1] eq "REG" && exists($argv{PROBE}) && _istrue($argv{PROBE})) {
			warning($self,"clprotocol: Probe for app $F[2] - NOT IMPLEMENTED YET");
		}
		#########################################################
		# If we are the master send the details to the other	#
		# nodes...						#
		#########################################################
		if($self->{MASTER} eq hostname) {
			queue_request($self,REQUEST=>"$req",TO => "??ALL??",PRIORITY => 1);
			msg($self,"clprotocol: Queued request to update remaining nodes.");
		} 
		return 0;
	}

	#################################################################################
	# The DEREG and APP_REMOVE will remove the definition of the application	#
	# from the local machine and if we are the master will send out the message	#
	# to the other nodes as well.. This is based on the code for the NODE_REMOVE.	#
	#################################################################################

	if($F[1] eq "DEREG" || $F[1] eq "APP_REMOVE") {
		my $action="R" if ($F[1] eq "DEREG" || $F[1] eq "APP_REMOVE");
		my $a=$argv{APP};
		msg($self,"clprotocol: Will attempt to remove application '$a' ...") if $action eq "R";
		msg($self,"clprotocol: Will attempt to add application '$a' ...") if $action eq "A";
		if(!exists($self->{app_status}->{$a}) && $action eq "R") {
			warning($self,"clprotocol: Application  '$a' not part of cluster - already removed?");
		}
		#########################################################
		# The node exists as part of the configuration, so 	#
		# remove from the local database and node_state tables.	#
		#########################################################
		if($action eq "R") {
			delete($self->{app_status}->{$a});
			my $db=new clstatdb($self->{statusdb});
			if(!defined($db)) {
				my ($error_rc,$error_msg)=clstatdb::error;
				error($self,"clprotocol: Unable to open connection to status database!");
				$sock->no_response;
				return 0;
			}
			$db->do("DELETE FROM app_status WHERE name = '$a';");
			$db->do("DELETE FROM app_registered WHERE name = '$a';");
			$db->disconnect; $db=undef;
		}
		#########################################################
		# If we are the master queue a request to all other	#
		# nodes...						#
		#########################################################

		if($self->{MASTER} eq hostname) {
			if($action eq "R") {
				queue_request($self,REQUEST=>"$req",TO => "??ALL??",PRIORITY => 1);
				msg($self,"clprotocol: Queued request to update remaining nodes.");
			}
		} 

		#########################################################
		# Now send reconfigure events to all clients, ignoring	#
		# return codes for now...				#
		#########################################################
		my @D=qw(statd hbd statd);
		my $daemon;
		reconfigure($self);
		for $daemon (@D) {
			my $conn=new daemon_connect(host => hostname,daemon=>"lha_$daemon", timeout => 3);
			if(!defined($conn)) {
				my $x=daemon_connect::error();
				error($self,"clprotocol: Unable to contact '$daemon' locally - no refresh sent!");
			} else {
				msg($self,"clprotocol: Sending 'RECONFIGURE' to '$daemon' locally.");
				$conn->send_request(MSG=>"RECONFIGURE",TIMEOUT=>2);
				$conn=undef;
			}
		}
		msg($self,"clprotocol: Application '$a' successfully removed from configuration.") if $action eq "R";
		msg($self,"clprotocol: Application '$a' successfully added to configuration.") if $action eq "A";
		$sock->send_response("OK");
		return 0;

	}

	#################################################################
	# APP_STATUS contains details of all apps that the master knows	#
	# about. 							#
	#################################################################
	if($F[1] eq "APP_STATUS") {
		my $db=new clstatdb($self->{statusdb});
		if(!defined($db)) {
			($error_rc,$error_msg)=clstatdb::error();
			cluck("clprotocol: Unable to open database [$error_msg]");
			$sock->no_response;	# One way request type
			return 0;
		}
		$db->create_if_needed("app_status");
		my @R=split(/\|/,$argv{APP_INFO});
		my $crec;
		for $crec (@R) {
			my @F=split(/,/,$crec);
			if(!$db->do("INSERT OR REPLACE INTO app_status (name,host,status,updatetime) VALUES ('$F[0]','$F[1]','$F[2]','$F[3]');")) {
				($error_rc,$error_msg)=clstatdb::error();
				cluck("clprotocol: Unable to update app_status table [$error_msg]");
			}
		}
		$sock->no_response;	# One way request type
		return 0;
	}
	if($F[1] eq "APP_UPDATE") {
		my $db=new clstatdb($self->{statusdb});
		if(!defined($db)) {
			my ($error_rc,$error_msg)=clstatdb::error();
			cluck("clprotocol: Unable to open database [$error_msg]");
			$sock->no_response;	# One way request type
			return 0;
		}
		$db->create_if_needed("app_status");
		#########################################################
		# If no update time has been specified, indicate now.	#
		#########################################################
		if(!exists($argv{UPDATETIME})) {
			$argv{UPDATETIME}=time;
		}
		if(exists($argv{VALID})) {
			my $n=join(",",sort(split(/,/,$argv{VALID})));
			if(!$db->do("INSERT OR REPLACE INTO app_status (name,host,status,updatetime,valid) VALUES ('$argv{APP}','$argv{NODE}','$argv{STATUS}','$argv{UPDATETIME}','$n');")) {
				($error_rc,$error_msg)=clstatdb::error;
				cluck("clprotocol: Unable to update app_status table [$error_msg]");
			}
		} else {
			if(!$db->do("INSERT OR REPLACE INTO app_status (name,host,status,updatetime) VALUES ('$argv{APP}','$argv{NODE}','$argv{STATUS}','$argv{UPDATETIME}');")) {
				($error_rc,$error_msg)=clstatdb::error;
				cluck("clprotocol: Unable to update app_status table [$error_msg]");
			}
		}
		$sock->no_response;	# One way request type

		#########################################################
		# Send cluster event is the application has stopped or 	#
		# started and this node is the master.			#
		#########################################################
		if($self->{MASTER} eq hostname) {
			my ($rc,$count,$warns);
			if($argv{STATUS} eq "RUNNING") {
				($rc,$count,$warns)=$self->{clevents}->send_msg(event=>"app.start",node=>$argv{NODE},app=>$argv{APP});
			}
			if($argv{STATUS} eq "STOPPED") {
				($rc,$count,$warns)=$self->{clevents}->send_msg(event=>"app.end",node=>$argv{NODE},app=>$argv{APP});
			}
			if(defined($rc)) {
				if($rc==-1) {
					error("clprotocol: $_") foreach(@$warns);
				} else {
					warning("clprotocol: $_") foreach(@$warns);
				}
			}
		}
		return 0;
	}

	if($F[1] eq "SET_MODDED_SETTINGS") {
		my $data=$argv{DATA};
		my @R=split(/\n/,pack("H*",$data));
		# Clear down existing records first if necessary...
		my $db=new clstatdb($self->{statusdb});
		if(!defined($db)) {
			warning("clprotocol: Got 'SET_MODDED_SETTINGS could not open status db - igorning!");
			$sock->no_response;
			return 0;
		}
		$db->create_if_needed("modded_settings");
		$db->do("DELETE FROM modded_settings;");
		my $crec;
		for $crec (@R) {
			my @Z=split(/\|/,$crec);
			my $i="INSERT INTO modded_settings (appname,attrname,value) VALUES ('$Z[0]','$Z[1]','$Z[2]');";
			$db->do($i);
		}
		$db->disconnect; $db=undef;
		$sock->no_response;
		return 0;
	}

	if($F[1] eq "GET_MODDED_SETTINGS") {
		if($self->{MASTER} ne hostname) {
			warning("clprotocol: Got 'GET_MODDED_SETTINGS - but not master - ignoring!");
			$sock->no_response;
			return 0;
		}
		my $req=$argv{REQUESTOR};
		# Get any modded_settings records...
		my @R=();
		my $db=new clstatdb($self->{statusdb});
		if(!defined($db)) {
			warning("clprotocol: Got 'GET_MODDED_SETTINGS could not open status db - igorning!");
			$sock->no_response;
			return 0;
		}
		$db->create_if_needed("modded_settings");
		my $hlist=$db->get_rows("SELECT * FROM modded_settings;");
		$db->disconnect; $db=undef;
		if(defined($hlist)) {
			for (@$hlist) {
				if(defined($_->{attrname}) && defined($_->{value})) {
					push @R,"$_->{appname}|$_->{attrname}|$_->{value}";
				}
			}
		}
		my $z=unpack("H*",join("\n",@R));
		queue_request($self,REQUEST=>"CLPROTO SET_MODDED_SETTINGS DATA=$z",TO => $req,PRIORITY => 5);
		$sock->no_response;
		return 0;
	}
	if($F[1] eq "VALID_NODES") {
		my $db=new cldb($self->{configdb});
		if(!defined($db)) {
			($error_rc,$error_msg)=cldb::error;
			cluck("clprotocol: Unable to open database: $error_msg");
			$sock->no_response;
			return 0;
		}

		my $href=$db->get_rows("SELECT name FROM nodes;");
		my @N=();
		if(!defined($href) || !defined($href->[0])) {
			error($self,"clprotocol: Port not found for 'lha_clusterd' in config database.",1);
		}
		for (@$href) {
			push @N,$_->{name};
		}
		$db->disconnect;

		#########################################################
		# Now connect to the status database.			#
		#########################################################

		$db=new clstatdb($self->{statusdb});
		if(!defined($db)) {
			my ($error_rc,$error_msg)=clstatdb::error();
			cluck("clprotocol: Unable to open database [$error_msg]");
			$sock->no_response;	# One way request type
			return 0;
		}
		my @N2=sort(split(/,/,$argv{NODES}));
		my $new_active_node_list=0;
		if(@N2 != @{$self->{SET_NODE_LIST}}) {
			$new_active_node_list=1;
		} else {
			my $x=0;
			while($x< @N2) {
				if($N2[$x] ne $self->{SET_NODE_LIST}->[$x]) {
					$new_active_node_list=1;
					last;
				}
				$x++;
			}
		}
		#########################################################
		# If we have a new active node list then schedule a 	#
		# send a one-way request to the local hbd daemon with	#
		# the new list of nodes...				#
		#########################################################
		if($new_active_node_list) {
			$self->{SET_NODE_LIST}=[@N2];
			my $x=join(",",@N2);
			msg($self,"clprotocol: Set_node_list changed to: $x");
			my $conn=new daemon_connect(host => hostname,daemon=>"lha_hbd", timeout => 2);
			if($conn) {
				$conn->send_request(MSG=>"SET_NODE_LIST|$x",TIMEOUT=>2);
			}
			$conn=undef;
		}

		foreach (@N2) {
			my $r=$db->do("UPDATE cluster_status_vars SET value = 'Joined' WHERE var = '$_:clusternodestate';");
			$self->{node_state}->{$_}="Joined";
			if(!$r) {
				my ($error_rc,$error_msg)=cldb::error;
				cluck("clprotocol: Unable to update cluster_status_vars table [$error_msg]");
			}
		}
		#########################################################
		# Now any that where in @N that are not in @N2 then	#
		# set the status to "Down".				#
		#########################################################

		my $cnode;
		for $cnode (@N) {
			if(!scalar(grep {$_ eq $cnode} @N2)) {
				my $r=$db->do("UPDATE cluster_status_vars SET value = 'Down' WHERE var = '$cnode:clusternodestate';");
				if(!$r) {
					my ($error_rc,$error_msg)=cldb::error;
					cluck("clprotocol: Unable to update cluster_status_vars table [$error_msg]");
				}
			}
		}
		$db->disconnect;
		$sock->no_response;
		return 0;
	}

	#################################################################
	# Node add and remove are quite similar so deal in one go.	#
	#################################################################
	if($F[1] eq "NODE_REMOVE" || $F[1] eq "NODE_ADD") {
		my $action="R" if $F[1] eq "NODE_REMOVE";
		$action="A" if $F[1] eq "NODE_ADD";
		my $n=$argv{NODE};
		msg($self,"clprotocol: Will attempt to remove node '$n' ...") if $action eq "R";
		msg($self,"clprotocol: Will attempt to add node '$n' ...") if $action eq "A";
		if(!exists($self->{node_state}->{$n}) && $action eq "R") {
			warning($self,"clprotocol: Node '$n' not part of cluster - already removed?");
		}
		#########################################################
		# The node exists as part of the configuration, so 	#
		# remove from the local database and node_state tables.	#
		#########################################################
		if($action eq "R") {
			my $db=new cldb($self->{configdb});
			if(!defined($db)) {
				($error_rc,$error_msg)=cldb::error;
				cluck("clprotocol: Unable to open database: $error_msg");
				error($self,"clprotocol: Node removal failed!");
				$sock->no_response;
				return 0;
			}
			$db->do("DELETE FROM nodes WHERE name = '$n';");
			$db->disconnect;
			$db=undef;
			delete($self->{node_state}->{$n});
			$db=new clstatdb($self->{statusdb});
			if(!defined($db)) {
				my ($error_rc,$error_msg)=clstatdb::error;
				error($self,"clprotocol: Unable to open connectino to status database!");
				$sock->no_response;
				return 0;
			}
			$db->do("DELETE FROM cluster_status_vars WHERE var = '$n:clusternodestate';");
			$db->disconnect; $db=undef;
		} else {
			$self->{node_state}->{$n}="No_response";
			my $db=new clstatdb($self->{statusdb});
			if(!defined($db)) {
				my ($error_rc,$error_msg)=clstatdb::error;
				error($self,"clprotocol: Unable to open connection to status database!");
				$sock->no_response;
				return 0;
			}
			$db->do("INSERT OR REPLACE INTO cluster_status_vars (var,value) VALUES('$n:clusternodestate','No_response');");
			$db->disconnect; $db=undef;
		}
		#########################################################
		# If we are the master queue a request to all other	#
		# nodes...						#
		#########################################################

		if($self->{MASTER} eq hostname) {
			if($action eq "R") {
				queue_request($self,REQUEST=>"CLPROTO NODE_REMOVE NODE=$n",TO => "??ALL??",PRIORITY => 1);
				msg($self,"clprotocol: Queued request to update remaining nodes.");
			} else {
				queue_request($self,REQUEST=>"CLPROTO NODE_ADD NODE=$n",TO => "??ALL??",PRIORITY => 1);
				msg($self,"clprotocol: Queued request to update remaining nodes.");
			}
		} 

		#########################################################
		# Now send reconfigure events to all clients, ignoring	#
		# return codes for now...				#
		#########################################################
		my @D=qw(statd hbd statd lockd);
		my $daemon;
		for $daemon (@D) {
			my $conn=new daemon_connect(host => hostname,daemon=>"lha_$daemon", timeout => 3);
			if(!defined($conn)) {
				my $x=daemon_connect::error();
				error($self,"clprotocol: Unable to contact '$daemon' locally - no refresh sent!");
			} else {
				if($daemon ne "hbd") {
					msg($self,"clprotocol: Sending 'RECONFIGURE' to '$daemon' locally.");
					$conn->send_request(MSG=>"RECONFIGURE",TIMEOUT=>4);
					$conn=undef;
				} else {
					if($action eq "R") {
						msg($self,"clprotocol: Sending 'REMOVE_NODE|$n' to '$daemon' locally.");
						$conn->send_request(MSG=>"REMOVE_NODE|$n",TIMEOUT=>3);
					} else {
						msg($self,"clprotocol: Sending 'ADD_NODE|$n' to '$daemon' locally.");
						$conn->send_request(MSG=>"ADD_NODE|$n",TIMEOUT=>3);
					}
					$conn=undef;
				}
			}
		}
		msg($self,"clprotocol: Node '$n' successfully removed from configuration.") if $action eq "R";
		msg($self,"clprotocol: Node '$n' successfully added to configuration.") if $action eq "A";
		$sock->send_response("OK");
		return 0;
	}
		
	#################################################################
	# Handle a message from the local heartbeat daemon - if we	#
	# are not the master send it onwards [oneway]. If the master	#
	# sends this message then mark the nodes as down then.		#
	#################################################################
	if($F[1] eq "NODE_FAILURE") {
		my $elective=0;
		my $source="?";
		if(exists($argv{ELECTIVE})) {
			$elective=int($argv{ELECTIVE});
		}
		if(exists($argv{SOURCE})) {
			$source=$argv{SOURCE};
			msg($self,"Source node for NODE_FAILURE message: $source.");
		}
		if($elective==0) {
			msg($self,"clprotocol: Got node failure message: $req");
		} else {
			msg($self,"clprotocol: Got elective node removal message: $req");
		}
		my $nodes=$argv{NODES};
		my $n;
		my $_changed=0;
		#########################################################################
		# If this is the master then send out node.down events for each		#
		# node in turn.								#
		#########################################################################
		if($self->{MASTER} eq hostname) {
			for $n (split(/,/,$nodes)) {
				my ($rc,$count,$warns);
				if($elective) {
					($rc,$count,$warns)=$self->{clevents}->send_msg(event=>"node.leave",node=>$n);
				} else {
					($rc,$count,$warns)=$self->{clevents}->send_msg(event=>"node.down",node=>$n);
				}
				if($rc==-1) {
					error("clprotocol: $_") foreach(@$warns);
				} else {
					warning("clprotocol: $_") foreach(@$warns);
				}
			}
		}
		#########################################################################
		# If the nodes include the master - then reform...			#
		#########################################################################
		for $n (split(/,/,$nodes)) {
			if(exists($self->{MASTER}) && $self->{MASTER} eq $n) {
				$sock->no_response;
				queue_request($self,REQUEST=>"CLPROTO REFORM DEAD=$nodes",TO => "??ALL??",PRIORITY => 1);
				msg($self,"Pushing out reform message to other nodes[2] ... dead=$nodes") if defined(&::msg);
				return (88,[],[split(/,/,$nodes)]);
			}
		}
		#########################################################################
		# Ignore the message if the master did not send it.			#
		#########################################################################
		if($elective==0) {
			if(exists($self->{MASTER}) && $source ne $self->{MASTER}) {
				$sock->no_response;
				return 0;
			}
		}

		#################################################################
		# Send a message to heartbeat daemons as soon as possible to	#
		# ensure it knows about the change to the host.			#
		#################################################################
		{
			my $conn=new daemon_connect(host => hostname,daemon=>"lha_hbd", timeout => 2);
			if(!defined($conn)) {
				my $x=daemon_connect::error();
				error($self,"clprotocol: Unable to contact 'hbd' locally - no refresh sent!");
			} else {
				msg($self,"clprotocol: Sending 'DEAD|$nodes' to 'hbd' locally.");
				$conn->send_request(MSG=>"DEAD|$nodes",TIMEOUT=>1);
				$conn=undef;
			}
		}

		#########################################################################
		# If it gets here it is probably from the master, so mark the node	#
		# as down...								#
		#########################################################################

		my @newdead=();
		if(exists($self->{MASTER}) && $self->{MASTER} ne hostname) {
			my $reform=0;
			for $n (split(/,/,$nodes)) {
				if($self->{node_state}->{$n} eq "Down") {
					msg($self,"clprotocol: Node already in state 'Down' - ignoring request.");
					next;
				}
				$_changed=1;
				$self->{node_state}->{$n}="Down";
				cluck("Unable to update status database!") if ! _setstatusvar($self,"$n:clusternodestate","Down");
				if($n eq $self->{MASTER}) {
					$reform=1;
				}
			}

			#################################################
			# At this point following the database update	#
			# we need to check the list of failed nodes.	#
			# If the list includes the current master, then	#
			# we've got to define a new master. In this	#
			# instance send out reform message and move	#
			# ourselves immediately to reform.		#
			#################################################

			if($reform) {
				msg($self,"clprotocol: Master died - attempt cluster reformation!");
				$sock->no_response;
				# queue_request($self,REQUEST=>"CLPROTO REFORM",TO => "??ALL??",PRIORITY => 1);
				# Create the restart_apps table if necessary

				# my $db=new clstatdb($self->{statusdb});
				# if(!defined($db)) {
					# my ($error_rc,$error_msg)=clstatdb::error;
					# cluck("Error opening status database:",$error_msg);
					# $sock->no_response;
					# return (88,());
				# }
				# if(! $db->create_if_needed("restart_apps")) {
					# $db->disconnect;
					# cluck("Unable to create 'restart_apps' table!");
					# $sock->no_response;
					# return (88,());
				# }
				# Query all applications that are 
				# my $q="SELECT * FROM app_status WHERE = 'RUNNING';";
				# my $hlist=$db->get_rows($q);
				# my @RR=();
				# if(defined($hlist)) {
					# for my $crec (@$hlist) {
						# if(defined($crec->{host}) && grep {$crec->{host} eq $_} @N) {
							# push @RR,$crec->{name};
						# }
					# }
				# }
				# So RR contains the list of apps that need to be
				# restarted after the cluster formation, so send
				# the list back to the cluster master daemon.
				# 
				msg($self,"Returning request to reform[3] ... dead=$nodes") if defined(&::msg);
				return (88,[],[split(/,/,$nodes)]);
			} else {
				if($elective) {
					$sock->no_response;
					return 0;
				}
				if($_changed) {
					msg($self,"Node failure detected [nodes=$nodes] - pushing message to master.");
					queue_request($self,REQUEST=>$req,TO => "??MASTER??",PRIORITY => 1);
				}
				$sock->no_response;
				return 0;
			}
		} else {
			for $n (split(/,/,$nodes)) {
				if($self->{node_state}->{$n} eq "Down") {
					msg($self,"clprotocol: Node already in state 'Down' - ignoring request.");
					next;
				}
				$_changed=1;
				$self->{node_state}->{$n}="Down";
				push @newdead,$n;
				cluck("Unable to update status database!") if ! _setstatusvar($self,"$n:clusternodestate","Down");
			}
		}
		if(exists($self->{MASTER}) && $self->{MASTER} eq hostname) {
			#################################################
			# We are the master so generate a list of apps	#
			# that need to be hosted elsewhere, and then	#
			# send a return code of 45 and the list of	#
			# apps.						#
			# Firstly though check the status of each node	#
			# listed and our idea of that node - if we	#
			# already have it down then discard it from the	#
			# list.						#
			#################################################
			if($elective) {
				my $cn;
				my @N=split(/,/,$argv{NODES});
				for $cn (@N) {
					$self->{node_state}->{$cn}="Down";
				}
				queue_request($self,REQUEST=>$req,TO => "??ALL??",PRIORITY => 1);
				$sock->no_response;
				return 0;
			}

			# for my $cn (split(/,/,$nodes)) {
			#		if(!exists($self->{node_state}->{$cn})) {
			#		cluck("node '$cn' not known - ignoring.");
			#		next;
			#	}
			#	if($self->{node_state}->{$cn} eq "Down") {
			#		cluck("Node failure already known for '$cn' - ignoring.") if $::DEBUG;
			#		next;
			#	}
			#	push @newdead,$cn;
			#	$self->{node_state}->{$cn}="Down";
			#	cluck("Unable to update status database!") if ! _setstatusvar($self,"$cn:clusternodestate","Down");
			#}
			msg($self,"clprotocol: newdead set to '" . join(",",@newdead) . "'.");

			#################################################
			# newdead contains the nodes that are currently	#
			# thought to be running so do something about	#
			# these.					#
			#################################################

			my $db=new clstatdb($self->{statusdb});
			if(!defined($db)) {
				my ($error_rc,$error_msg)=clstatdb::error;
				cluck("Error opening status database:",$error_msg);
				$sock->no_response;
				return 0;
			}
			if(! $db->create_if_needed("app_status")) {
				$db->disconnect;
				cluck("Unable to create 'app_status' table!");
				$sock->no_response;
				return 0;
			}
			
			my $n;
			#################################################
			# Query the app_status table for any running 	#
			# apps that are any of the newdead nodes,	#
			# and returning the information back to the	#
			# cluster deamon to handle.			#
			#################################################
			my @RR=();
			for $n (@newdead) {
				my $q="SELECT * FROM app_status WHERE host = '$n' AND status != 'STOPPED';";
				msg($self,"clprotocol: Query: $q");
				my $hlist=$db->get_rows($q);
				if(defined($hlist)) {
					for (@$hlist) {
						if(defined($_->{host}) && defined($_->{name})) {
							push @RR,"$n|$_->{name}";
						}
					}
				}
			}
			msg($self,"clprotocol: RR set to '" . join(",",@RR) . "'.");
			return (45,@RR);
		}
		$sock->no_response;
		return 0;
	}

	if($F[1] eq "WILLING_TO_FORM") {
		my $node=$F[2];
		# Only respond if currently the cluster master.
		if(!exists($self->{MASTER})) {
			msg($self,"clprotocol: Ignoring WILLING_TO_FORM [no master defined yet]");
			return 0;
		}

		if($self->{MASTER} ne hostname) {
			if($node ne $self->{MASTER}) {
				msg($self,"clprotocol: Ignoring WILLING_TO_FORM [not master]");
				$sock->no_response;
				return 0;
			} else {
				#################################################
				# The master does not realise it is the master!	#
				# So send it a master to indicate it is!	#
				#################################################
				queue_request($self,REQUEST=>"CLPROTO YOU_ARE_MASTER " . hostname,
							TO => $node, PRIORITY => 1);
				$sock->no_response;
				return;
			}
		}
		msg($self,"clprotocol: Sending back FOLLOW_ME [I am master].");
		queue_request($self,REQUEST=>"CLPROTO FOLLOW_ME " . hostname,
					TO => $node, PRIORITY => 1);
		$sock->no_response;
		return 0;
	}
	if($F[1] eq "FOLLOWER") {
		# Only respond if currently the cluster master.
		if(!exists($self->{MASTER}) || $self->{MASTER} ne hostname) {
			return 0;
		}
		# We need to tell the other nodes that the new node has
		# joined too...
		my $node=$F[2];
		$self->{node_state}->{$node}="Joined";
		cluck("Unable to update status database!") if ! _setstatusvar($self,"$node:clusternodestate","Joined");
		queue_request($self,REQUEST=>"CLPROTO ADD_NODE NODE=$node",
					TO => "??ALL??", PRIORITY => 1);
		msg($self,"");
		msg($self,"*** NODE JOINED CLUSTER - $node ***");
		msg($self,"");
		$sock->no_response;
		# Send out node.join message.
		my ($rc,$count,$warns)=$self->{clevents}->send_msg(event=>"node.join",node=>$node);
		if($rc==-1) {
			error("clprotocol: $_") foreach(@$warns);
		} else {
			warning("clprotocol: $_") foreach(@$warns);
		}
		return 0;
	}
	if($F[1] eq "FOLLOW_ME") {
		my $node=$F[2];
		$self->{node_state}->{$node}="Joined";
		$self->{MASTER}=$node;
		my $n=hostname;
		cluck("Unable to update status database!") if ! _setstatusvar($self,"$n:clusternodestate","Joined");
		cluck("Unable to update status database!") if ! _setstatusvar($self,"master",$node);
		queue_request($self,REQUEST=>"CLPROTO FOLLOWER " . hostname,
					TO => $node, PRIORITY => 5);
		return 0;
	}
	$sock->send_response("ERROR '$F[1]' request not recognised'");
}

#########################################################################
# Send a message to the master - though does nothing if the local	#
# machine is already the master!					#
#########################################################################

sub send_to_master {
my $self=shift;

	return("AM_MASTER") if $self->{MASTER} eq hostname;
}


sub active_nodes {
my $self=shift;

	my $c=0;
	for (keys %{$self->{node_state}}) {
		$c++ if $self->{node_state}->{$_} eq "Joined";
	}
	return $c;
}

sub active_node_list {
my $self=shift;
my @L;
	for (keys %{$self->{node_state}}) {
		push @L,$_ if $self->{node_state}->{$_} eq "Joined";
	}
	return (@L);
}

sub get_server_port {
my $self=shift;

	return $self->{SERVER_SOCKET};
}

#########################################################################
#									#
# There will be occasions when the protocol must send information out	#
# to other nodes, and when this is required, for example when it has	#
# joined the cluster and wants to send FOLLOWER following formation,	#
# then it adds it to the list maintained here. When a request is to	#
# be sent it must be queued and it can provide the following arguments:	#
# REQUEST	- What to send						#
# TO		- Who to - can be '??ALL??' - for all nodes.		#
#		                  '??MASTER??' - to cluster master.	#
#		                  n[,...] - just the nodes specified.	#
# ACK           - When a response is required [i.e. not a one-way	#
#		  request] then this should be set to the ref of a 	#
#		  sub to call [passed request,name,result,tag] for each	#
#		  ack. - OPTIONAL.					#
# TAG 		- An arbitary string to pass back to the ack routine -	# 
#		  to help it identify what to do.			#
# PRIORITY      - Since messages can 'queue' a priority can be assigned	#
#		  to re-order messages. The default priority is 10,	#
#		  with lower numbers being sent earlier.		#
#									#
# It should also be pointed out that duplicate entries are discarded	#
# every time the list is processed to reduce traffic.			#
#									#
#########################################################################

{
my @REQUESTS=();

sub queue_request {
my $self=shift;
my %args=@_;

	if(!exists($args{REQUEST}) || !exists($args{TO})) {
		debug($self,"clprotocol: Request not queued - missing args!!!");
		return -1; # Not queued - missing argument.
	}
	my $item={};
	$item->{REQUEST}=$args{REQUEST};
	$item->{TO}=$args{TO};
	if(exists($args{ACK})) {
		$item->{ACK}=$args{ACK};
	}
	if(exists($args{PRIORITY})) {
		$item->{PRIORITY}=$args{PRIORITY};
	} else {
		$item->{PRIORITY}=10;
	}
	push @REQUESTS,$item;
	debug($self,"clprotocol: Queued request '$args{REQUEST}'.");
	return 0;
}

sub ismaster {
my $self=shift;

	return 1 if hostname eq $self->{MASTER};
}

#########################################################################
# Passed a reference to an item perform the relevant work on that item.	#
#########################################################################

sub _handle_item {
my $self=shift;
my $item=shift;

	my $to=$item->{TO};
	my $request=$item->{REQUEST};
	debug($self,"clprotocol: Processing request '$request'."); 
	my $ack=undef; my $tag=undef;
	if(exists($item->{ACK})) {
		$ack=$item->{ACK};
	}
	if(exists($item->{TAG})) {
		$tag=$item->{TAG};
	}

	my @N=();
	if($to eq "??MASTER??") {
		@N=($self->{MASTER});
	} elsif($to eq "??ALL??") {
		@N=();
		my $chost;
		for $chost (keys(%{$self->{node_state}})) {
			next if $chost eq hostname; # ignore ourselves
			next if $self->{node_state}->{$chost} ne "Joined"; # ignore non-live
			push @N,$chost;
		}
	} else {
		@N=split(/,/,$to);
	}

	my $chost;
	for $chost (@N) {
		if(!exists($self->{channel}->{$chost})) {
			my $cluster_port=$self->{cluster_port};
			my $ips=$self->{nodes}->{$chost};
			if(!defined($ips)) {
				warning($self,"No IP details defined for '$chost'!");
				next;
			}
			my $PKI=$self->{pki};
			debug($self,"Sending '$request' to '$ips'.");
			my $channel=new C_client2(PKI=>$PKI,SERVER_INET=>$ips,SERVER_PORT=>$cluster_port,SERVER_NAME=>$chost);
			if(!defined($channel)) {
				warning($self,"Unable to create a channel to send a request to '$chost'!");
			}
			$self->{channel}->{$chost}=$channel;
		}
		if(defined($self->{channel}->{$chost})) {
			if(defined($ack)) {
				my $r=$self->{channel}->{$chost}->send_request($request);
				debug($self,"Sending '$request' to '$chost' [expecting ack].");
				&$ack($self,$request,$chost,$r,$tag);
			} else {
				debug($self,"Sending '$request' to '$chost' [no ack].");
				$self->{channel}->{$chost}->send_request($request,1);
			}
		}
	}
}

sub process_queue {
my $self=shift;
my $maxitems=shift;

	$maxitems=99 if !defined($maxitems);
	my $had=0;
	@REQUESTS=sort {$a->{PRIORITY} <=> $b->{PRIORITY};} @REQUESTS;
	debug($self,"clprotocol: process_queue - about to loop...");
	while(@REQUESTS && $had<$maxitems) {
		my $req=shift(@REQUESTS);
		debug($self,"clprotocol: process_queue: " . Dumper($req));
		_handle_item($self,$req);
		$had++;
	}
	return $had;
}

}

#########################################################################
# This takes two args DEAD_NODE and APP - and queries teh list of	#
# valid node and return the name of a node to use.			#
# It will use the availnodes and nextnode routines from the clutils	#
# library to handle the complexities of choosing the next node and also	#
# whether a fail-over is possible at all.				#
#									#
# it will return (res,node) - where;					#
# (1,node) - ok to start it on 'node'					#
#	'node' might be:						#
#	* $$NONE - no suitable node available.				#
#	* $$NOFAIL - failover disabled for application.			#
#	* $$DEAD_NODE n - node n specified is as next node is dead.	#
#									#
# (-1,msg) - an unexpected error [see 'msg'] has occured.		#
#									#
#########################################################################

sub get_alternative_node {
my $self=shift;
my %args=@_;

	#################################################################
	# We work out the valid nodes associated with the app, and	#
	# remove DEAD_NODE from the list if it is still in it.		#
	#################################################################

	if(!exists($args{DEAD_NODE}) || !exists($args{APP})) {
		return(-1,"Missing mandatory argument");
	}
	my ($app,$node)=($args{APP},$args{DEAD_NODE});

	my $db=new clstatdb($self->{statusdb});
	if(!defined($db)) {
		my ($error_rc,$error_msg)=clstatdb::error;
		return (-1,$error_msg);
	}

	my @r=nextnode($app);
	if(!defined($r[0])) {
		return(-1,$r[1]);
	}
	return(1,$r[0]);
}

sub app_status {
my $self=shift;
my $details=$_[0];
my @f=split(/\|/,$details);

	$self->{app_status}={};
	for (@f) {
		my @c=split(/,/,$_);
		$self->{app_status}->{$c[0]}={host=>$c[1],status=>$c[2],updatetime=>$c[3]};
	}
}

#########################################################################
# This is called by cluster daemon on the master to start a particular	#
# application on a particular node. It is expected that the following	#
# arguments are passed via a hash:					#
# DEAD_NODE	- The node that has died [and where the app was running]#
# NEW_NODE	- The new node to run the application on		#
# APP		- The application to start				#
#									#
# This routine will locally run a program [in the background# to start	#
# the application in question on the node. Prior to that it will remove	#
# the dead node from the list of valid nodes for the application.	#
#									#
# Note:	Since this routine might be called several times there is a 	#
#	possibility of two or more applications starting at the same	#
#	time. Hence the 'lha_startapp' should ensure adequate resource	#
#	locking is in place - luckily this is almost inherent in all	#
#	places because most activity takes place via network sockets	#
#	or locking-friendly access to the status or configuration 	#
#	databases.							#
#									#
# Return Codes:								#
# 0 - failed to start application					#
# 1,NODE,PID - initialised the application fail-over			#
#	  PID is the process ID on the remote node - not locally!	#
#########################################################################

sub start_on_node {
my $self=shift;
my %args=@_;

	my $chost;
	my @N=();
	for $chost (keys(%{$self->{node_state}})) {
		next if $self->{node_state}->{$chost} ne "Joined"; # ignore non-live
		push @N,$chost;
	}
	my $clreqdxml=new clreqdxml("$::ENV{TRUECL_CFG}/clreqd.xml");
	if(! defined($clreqdxml)) {
		error("Unable to load Request Daemon configuration info.", "Reason: " . clreqdxml::error(),1);
		return 0;
	}

	msg($self,"N=" . join(",",@N));
	my $reqd=new clreq(hosts => [@N], clreqdxml => $clreqdxml, logger=>$::LOGGER, timeout=>3);
	my $r=$reqd->run(cmd => ["perform_adhoc_sql","DBTYPE","status","SQL","UPDATE app_status SET status = 'STOPPED', host = '' WHERE name = '$args{APP}';"]);
	msg($self,"Results of forcing app status to down:");
	msg($self,Dumper($r));

	#################################################################
	# Now the list of valid nodes for the application have been	#
	# changed across the cluster we can now attempt to start the	#
	# application on the remote node...				#
	#################################################################

	my $ok=join(",",@N);
	$reqd=new clreq(hosts => [$args{NEW_NODE}],clreqdxml => $clreqdxml, logger=>$::LOGGER, timeout=>6);
	$r=$reqd->run(cmd => ["complete_start_app","APP",$args{APP},"OK",$ok]);
	if(!defined($r)) {
		error("The 'complete_start_app' failed to run to start the application '$args{APP}'.");
		return 0;
	}
	if($r->{$args{NEW_NODE}} !~ /^OK\s+(\d+)/) {
		error($args{NEW_NODE});
		return 0;
	}
	my $ppp=$1;
	msg($self,"Successfully initiated start of application '$args{APP}' on node '$args{NEW_NODE}' process id '$ppp'.");
	return (1,$args{NEW_NODE},$ppp);
}

1;

